package cn.cs.message.consumer;

import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;

import cn.cs.message.Message;
import cn.cs.message.ReconnectHandler;

public abstract class MessageConsumerBatch extends MessageConsumer
{
	private static final Logger logger=LoggerFactory.getLogger(MessageConsumerBatch.class);
	private static final long NEXTDELEVER_TIMEOUT=1000l;	
	
	protected ExecutorService threadPool;
	protected Map<Thread, Channel> channelManager;
	
	@Override
	public void doStart() throws IOException, TimeoutException
	{
		//启动线程池
		channelManager=new ConcurrentHashMap<Thread, Channel>();
		threadPool=new ThreadPoolExecutor(threadPoolSize, threadPoolSize,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(),
                new ThreadFactory(){
			@Override
			public Thread newThread(Runnable r) 
			{
				Thread thread=new Thread(r);
				try {
					Channel channel = connectManage.createChannel();
					if(channel!=null)
					{
						channelManager.put(thread, channel);
						channel.basicQos(qos);
					}
				} catch (IOException e) {
					logger.warn(e.getMessage(),e);					
				}
				return thread;
			}			
		}){
			 protected void afterExecute(Runnable r, Throwable t) {
				 super.afterExecute(r, t);
				 if(t!=null)
				 {
					 logger.error("Throwable",t);	
					//关闭，删除channel
					 Thread thread = Thread.currentThread();
					 Channel channel = channelManager.get(thread);
					 try {
						channel.close();
					} catch (IOException e) {
						e.printStackTrace();
					} catch (TimeoutException e) {
						e.printStackTrace();
					}finally{
						channelManager.remove(thread);
					}
				 }
			 }
		};
		
		for(int i=0;i<threadPoolSize;i++)
			threadPool.execute(getRunable());
	}
	
	protected Runnable getRunable() 
	{
		return new Runnable() {
			private int count=0;
			private volatile boolean isRun;
			private Map<Long, Message> msgMap=new HashMap<Long,Message>();//
			private Map<Long, Message> msgReMap=new HashMap<Long,Message>();//
			private Channel channelCopy;	
			private QueueingConsumer consumer;
			@Override
			public void run() 
			{
				while(true)
				{
					QueueingConsumer tempConsumer=null;
					Channel tempChannel=null;
					try {
						Thread thread=Thread.currentThread();
						tempChannel=channelManager.get(thread);
						if(tempChannel!=null)
						{
							if(tempChannel.isOpen())
								tempChannel.close();
							channelManager.remove(thread);
						}
						tempChannel=connectManage.createChannel();
						final Channel channel=tempChannel;
						if(tempChannel!=null)
						{	
							tempChannel.basicQos(qos);
							channelManager.put(thread, tempChannel);
						}	
						boolean autoAck=false;			
						tempConsumer= new QueueingConsumer(channel){
							@Override
						    public void handleCancel(String consumerTag) throws IOException {
								reminder.remind("消费的队列可能丢失");
								isRun=false;
								threadPool.execute(getRunable());
							}
						};
						channel.basicConsume(messageNameService.getQueueName(appCode), autoAck, tempConsumer);
						channelCopy=tempChannel;
						consumer=tempConsumer;
						loopRun();
						break;
					}catch (Exception e) {
						logger.warn("等待1分钟后再尝试建立consumer和队列的连接",e);
					}
					if(channelCopy==null||consumer==null)
						try {
							TimeUnit.SECONDS.sleep(60);
						} catch (InterruptedException e1) {
							e1.printStackTrace();
						}	
				}
			}
			
			private void loopRun()
			{
				isRun=true;
				while (isRun) 
				{
					try {
						QueueingConsumer.Delivery delivery = consumer.nextDelivery(NEXTDELEVER_TIMEOUT);						
						//	暂停则批处理一次
						if(delivery==null&&(msgMap.size()>0||msgReMap.size()>0))
							batchHandler();
						else if(delivery!=null) //接收放置到map里 
						{
							Message rmqMsg=getFromBytes(delivery.getBody());					
							if(delivery.getEnvelope().isRedeliver())
								msgReMap.put(delivery.getEnvelope().getDeliveryTag(), rmqMsg);
							else
								msgMap.put(delivery.getEnvelope().getDeliveryTag(), rmqMsg);
							count++;
							
							//	如果到达qos则批处理一次
							if(count==qos)
								batchHandler();
						}
					} catch (ShutdownSignalException e1) {
						logger.warn("可能rabbitMQ服务器挂掉！",e1);								
						return;								
					} catch (ConsumerCancelledException e1) {
						logger.warn("已被覆盖，不可能到这里！",e1);								
						return;	
					} catch (InterruptedException e1) {
						logger.warn("线程中断！",e1);	
						return;	
					} catch (Exception e){
						logger.error("未预见的错误",e);
					}
				}
			}
			
			private void batchHandler() throws IOException 
			{
				try {
					if(msgMap.size()>0)
						handleMessage(msgMap.values());
				
					if(msgReMap.size()>0)
						handleMessageRe(msgReMap.values());
					
				} catch (Exception e){
					logger.error("未预见的错误",e);
				} 
				for(Long tag:msgMap.keySet())
					channelCopy.basicAck(tag, false);
				for(Long tag:msgReMap.keySet())
					channelCopy.basicAck(tag, false);
				//清空
				count=0;
				msgMap.clear();
				msgReMap.clear();
			}
			
		};	
	}
	
	public abstract void handleMessage(Collection<Message> msgs);
	
	public abstract void handleMessageRe(Collection<Message> msgs);
	
	@Override
	public void doStop() throws IOException, TimeoutException
	{
		Collection<Channel> channels=channelManager.values();
		for(Channel temp:channels)
			temp.close();
		
		if(threadPool!=null)
			threadPool.shutdown();
	}
	
	@Override
	public ReconnectHandler getReconnectHandler() {
		if(reconnectHandler==null)
		{
			reconnectHandler=new ReconnectHandler() 
			{
				@Override
				public void afterReconnect() {
					for(int i=0;i<threadPoolSize;i++)
						threadPool.execute(getRunable());
				}
			};
		}	
		return reconnectHandler;
	}

	
}
